
#include <gst/gst.h>
#include <gst/base/gstadapter.h>

#include "gstmpegdefs.h"

#define GST_TYPE_H222TS_DEMUX \
  (gst_h222ts_demux_get_type())
#define GST_H222TS_DEMUX(obj) \
  (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_H222TS_DEMUX, GstH222TSDemux))
#define GST_H222TS_DEMUX_CLASS(klass) \
  (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_H222TS_DEMUX, GstH222TSDemuxClass))
#define GST_H222TS_DEMUX_GET_CLASS(obj) \
  (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_H222TS_DEMUX, GstH222TSDemuxClass))

#define GST_H222TS_MIME_TYPE_ES "application/x-mpegts-elementary-stream"
#define GST_H222TS_PROGRAM_MAX 0x10000
#define GST_H222TS_PID_MAX 0x200

typedef struct _GstH222TSDemuxClass GstH222TSDemuxClass;
typedef struct _GstH222TSDemux GstH222TSDemux;

struct _GstH222TSDemuxClass
{
  GstElementClass parent_class;
  GstPadTemplate *tmpl_sink;
  GstPadTemplate *tmpl_src;

  /* v-method */
    GstFlowReturn (*take_transport_packet_buffer) (GstAdapter * in,
      GstBuffer ** out);
};

struct _H222TSProgram
{
  guint program_number;
  guint16 PCR_PID;
  guint16 PMT_PID;
  guint64 PCR;
};
typedef struct _H222TSProgram H222TSProgram;

struct _GstH222TSDemux
{
  GstElement element;
  GstPad *sinkpad;
  GstAdapter *adapter;
  guint16 network_PID;
  H222TSProgram program;
  void **pidtab;
};

GType gst_h222ts_demux_get_type (void);

/************************************************************************************
 * Private Code
 */
#include <string.h>

#define H222TS_PROGRAM(demux, n) \
  ((demux)->program.program_number == (n) ? &(demux)->program : NULL)
#define H222TS_PID_CONTEXT(demux, pid) \
  ((H222TSPIDContext*)(demux)->pidtab[pid])
#define H222TS_PID_CONTEXT_SET(demux, pid, val) \
  ((demux)->pidtab[pid] = (val))
#define H222TS_SECTION(demux, pid) \
  ((H222TSSection*)H222TS_PID_CONTEXT ((demux), (pid)))
#define H222TS_STREAM(demux, pid) \
  ((H222TSStream*)H222TS_PID_CONTEXT ((demux), (pid)))

#define g_memset(a,b,c) memset((a),(b),(c))

#define H222TS_NORMAL_TS_PACKETSIZE 188
#define SECTION_HEADER_LENGTH 8

typedef struct _H222TSPIDContext H222TSPIDContext;
typedef struct _H222TSSection H222TSSection;
typedef struct _H222TSStream H222TSStream;
typedef struct _H222TSPacket H222TSPacket;
typedef struct _H222TSPESPacket H222TSPESPacket;
typedef struct _H222TSSectionHeader H222TSSectionHeader;

typedef GstFlowReturn (*H222TSPacketProcess) (GstH222TSDemux *, H222TSPacket *);

struct _H222TSPIDContext
{
  gint program_number;
  gint8 continuity_counter;

  GstAdapter *adapter;
  H222TSPacketProcess process;
};

struct _H222TSPacket
{
  /* transport_packet() */
  guint8 transport_error_indicator;
  guint8 payload_unit_start_indicator;
  guint16 PID;
  guint8 transport_priority;
  guint8 transport_scrambling_control;
  guint8 adaptation_field_control;
  guint8 continuity_counter;

  /* if adaptation_field_control == '10' || adaptation_field_control == '11' */
  guint8 discontinuity_indicator;
  guint8 random_access_indicator;
  guint8 elementary_stream_priority_indicator;
  guint8 PCR_flag;

  guint8 OPCR_flag;
  guint8 splicing_point_flag;
  guint8 transport_private_data_flag;
  guint8 ltw_valid_flag;

  guint8 splice_countdown;
  guint8 splice_type;
  guint16 ltw_offset;
  guint piecewise_rate;
  guint64 PCR;
  guint64 OPCR;
  guint64 DTS_next_AU;
  GstBuffer *private_data_buffer;

  /* payload */
  GstBuffer *data_buffer;
};

struct _H222TSPESPacket
{
  guint packet_start_code_prefix;
  guint16 PES_packet_length;
  guint8 stream_id;
  guint8 PES_scrambling_control;
  guint8 PES_priority;
  guint8 data_alignment_indicator;
  guint8 copyright;
  guint8 original_or_copy;
  guint8 PTS_DTS_flags;
  guint8 ESCR_flag;
  guint8 ES_rate_flag;
  guint8 DSM_trick_mode_flag;
  guint8 additional_copy_info_flag;
  guint8 PES_CRC_flag;
  guint8 PES_extension_flag;
  guint8 PES_header_data_length;

  /* if PTS exists */
  guint64 PTS;
  /* if DTS exists */
  guint64 DTS;
  /* if ESCR_flag */
  guint64 ESCR;
  /* if ES_rate_flag */
  guint ES_rate;
  /* if DSM_trick_mode_flag */
  guint8 trick_mode_control;
  /* if additional_copy_info_flag */
  guint8 additional_copy_info;
  /* if PES_CRC_flag */
  guint16 previous_PES_packet_CRC;
  /* if PES_extension_flag */
  guint8 PES_private_data_flag;
  guint8 pack_header_field_flag;
  guint8 program_packet_sequence_counter_flag;
  guint8 P_STD_buffer_flag;
  guint8 PES_extension_flag_2;

  /* if program_packet_sequence_counter_flag */
  guint8 program_packet_sequence_counter;
  guint8 MPEG1_MPEG2_indicator;
  guint8 original_stuff_length;
  /* if P_STD_buffer_flag */
  guint16 P_STD_buffer_size;
  guint8 P_STD_buffer_scale;

  guint8 dummy_for_align;

  GstBuffer *PES_packet_data_buffer;
};

struct _H222TSSectionHeader
{
  guint8 table_id;
  guint8 section_syntax_indicator;
  guint16 section_length;
  guint16 program_number;
  guint8 version_number;
  guint8 current_next_indicator;
  guint8 section_number;
  guint8 last_section_number;
};

struct _H222TSSection
{
  H222TSPIDContext pid_context;

  H222TSSectionHeader header;
};

struct _H222TSStream
{
  H222TSPIDContext pid_context;
  GstPad *pad;
  guint8 stream_type;
};

/*
 * prototypes
 */
static GstFlowReturn h222ts_demux_take_transport_packet_buffer (GstAdapter *,
    GstBuffer **);
static GstFlowReturn h222ts_demux_sink_chain (GstPad * pad, GstBuffer * buffer);
static H222TSSection *h222ts_demux_section_new (void);
static GstFlowReturn h222ts_demux_process_section (GstH222TSDemux * demux,
    H222TSPacket * packet);
static GstFlowReturn h222ts_demux_process_stream (GstH222TSDemux * demux,
    H222TSPacket * packet);

/*
 * static variables
 */
static GstStaticPadTemplate s_tmpl_sink = GST_STATIC_PAD_TEMPLATE ("sink",
    GST_PAD_SINK,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS ("video/mpegts")
    );

static GstStaticPadTemplate s_tmpl_src = GST_STATIC_PAD_TEMPLATE ("src_%04X",
    GST_PAD_SRC,
    GST_PAD_SOMETIMES,
    GST_STATIC_CAPS (GST_H222TS_MIME_TYPE_ES ", "
        "stream_type = (int) [ 1, 255 ]; "
        "video/mpeg, "
        "mpegversion = (int) { 1, 2, 4 }, "
        "systemstream = (boolean) FALSE; "
        "video/x-h264; "
        "audio/mpeg, "
        "mpegversion = (int) { 1, 4 }; "
        "audio/x-lpcm; " "audio/x-ac3; " "aduio/x-dts; "));

gboolean
gst_h222ts_demux_plugin_init (GstPlugin * plugin)
{
  if (!gst_element_register (plugin, "h222tsdemux",
          GST_RANK_PRIMARY, GST_TYPE_H222TS_DEMUX))
    return FALSE;

  return TRUE;
}

static void
h222ts_demux_base_init (GstH222TSDemuxClass * klass)
{
  GstElementClass *element_class = GST_ELEMENT_CLASS (klass);

  klass->tmpl_sink = gst_static_pad_template_get (&s_tmpl_sink);
  klass->tmpl_src = gst_static_pad_template_get (&s_tmpl_src);

  gst_element_class_add_pad_template (element_class, klass->tmpl_sink);
  gst_element_class_add_pad_template (element_class, klass->tmpl_src);

  gst_element_class_set_details_simple (element_class,
      "TEXT1", "TEXT2", "TEXT3", "Tetsuya Yasuda <tetsuyayasuda@gmail.com>");

  klass->take_transport_packet_buffer =
      h222ts_demux_take_transport_packet_buffer;
}

static void
h222ts_demux_class_init (GstH222TSDemuxClass * klass)
{
}

static void
h222ts_demux_init (GstH222TSDemux * demux)
{
  GstH222TSDemuxClass *klass = GST_H222TS_DEMUX_GET_CLASS (demux);

  demux->sinkpad = gst_pad_new_from_template (klass->tmpl_sink, "sink");
  gst_pad_set_chain_function (demux->sinkpad, h222ts_demux_sink_chain);
  gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);

  demux->adapter = gst_adapter_new ();

  g_memset (&(demux->program), 0, sizeof (H222TSProgram));
  demux->program.program_number = -1;

  demux->pidtab = g_malloc0 (GST_H222TS_PID_MAX * sizeof (void *));
  H222TS_PID_CONTEXT_SET (demux, PID_PROGRAM_ASSOCIATION_TABLE,
      h222ts_demux_section_new ());

  GST_DEBUG_OBJECT (demux, "GstH222TSDemux initialized");
}

GType
gst_h222ts_demux_get_type (void)
{
  static GType h222ts_demux_type = 0;

  if (G_UNLIKELY (!h222ts_demux_type)) {
    static const GTypeInfo h222ts_demux_info = {
      sizeof (GstH222TSDemuxClass),
      (GBaseInitFunc) h222ts_demux_base_init,
      NULL,
      (GClassInitFunc) h222ts_demux_class_init,
      NULL,
      NULL,
      sizeof (GstH222TSDemux),
      0,
      (GInstanceInitFunc) h222ts_demux_init,
    };

    h222ts_demux_type = g_type_register_static
        (GST_TYPE_ELEMENT, "GstH222TSDemux", &h222ts_demux_info, 0);
  }

  return h222ts_demux_type;
}

static void
h222ts_demux_pid_context_init (H222TSPIDContext * ctx,
    H222TSPacketProcess process)
{
  ctx->adapter = gst_adapter_new ();
  ctx->process = process;
  ctx->program_number = -1;
  ctx->continuity_counter = -1;
}

static H222TSSection *
h222ts_demux_section_new (void)
{
  H222TSSection *section = g_malloc0 (sizeof (H222TSSection));
  h222ts_demux_pid_context_init
      (&section->pid_context, h222ts_demux_process_section);
  return section;
}

static H222TSStream *
h222ts_demux_stream_new (void)
{
  H222TSStream *stream = g_malloc0 (sizeof (H222TSStream));
  h222ts_demux_pid_context_init
      (&stream->pid_context, h222ts_demux_process_stream);
  return stream;
}

#if 0
static H222TSProgram *
h222ts_demux_program_new (void)
{
  H222TSProgram *program = g_malloc0 (sizeof (H222TSProgram));
  program->PCR_PID = GST_H222TS_PID_MAX;
  program->PMT_PID = GST_H222TS_PID_MAX;
  return program;
}

static void
h222ts_demux_program_free (H222TSProgram * program)
{
  g_free (program);
}

static void
h222ts_demux_stream_free (H222TSStream * stream)
{
  g_object_unref (stream->pid_context.adapter);
  g_free (stream);
}

static void
h222ts_demux_section_free (H222TSSection * section)
{
  g_object_unref (section->pid_context.adapter);
  g_free (section);
}
#endif

static guint64
h222ts_demux_read_PCR (const guint8 * data)
{
  guint32 pcr1;
  guint16 pcr2;
  guint64 pcr, pcr_ext;

  pcr1 = GST_READ_UINT32_BE (data);
  pcr2 = GST_READ_UINT16_BE (data + 4);
  pcr = ((guint64) pcr1) << 1;
  pcr |= (pcr2 & 0x8000) >> 15;
  pcr_ext = (pcr2 & 0x01ff);
  if (pcr_ext)
    pcr = (pcr * 300 + pcr_ext % 300) / 300;
  return pcr;
}

static GstFlowReturn
h222ts_demux_has_enough_data_for_PES_packet (GstAdapter * adapter)
{
  const guint8 *data;
  guint16 length;

  data = gst_adapter_peek (adapter, 6);
  if (G_UNLIKELY (data == NULL))
    goto more_data;

  if (G_UNLIKELY (data[0] != 0 || data[1] != 0 || data[2] != 1)) {
    GST_DEBUG_OBJECT (adapter, "lost sync");
    return GST_FLOW_LOST_SYNC;
  }

  length = (data[4] << 8) + data[5];

  if (length == 0)
    goto more_data;

  if (G_UNLIKELY (length + 6 > gst_adapter_available (adapter)))
    goto more_data;

  return GST_FLOW_OK;

more_data:
  GST_DEBUG_OBJECT (adapter, "need more data");
  return GST_FLOW_NEED_MORE_DATA;
}

static GstFlowReturn
h222ts_demux_parse_PES_packet (GstAdapter * adapter, H222TSPESPacket * packet)
{
  guint stream_id, offset = 0;
  guint PES_packet_length, avail;
  const guint8 *data;

  g_memset (packet, 0, sizeof (H222TSPESPacket));

  /*
   * assume that adapter has enough amount of data for PES packet,
   * and that sync is gone.
   */
  data = gst_adapter_peek (adapter, 6);

  if (G_UNLIKELY (data == NULL))
    goto invalid_length;

  stream_id = packet->stream_id = data[3];
  PES_packet_length = packet->PES_packet_length = (data[4] << 8) + data[5];

  if (PES_packet_length == 0)
    avail = gst_adapter_available (adapter);
  else
    avail = 6 + PES_packet_length;

  data = gst_adapter_peek (adapter, avail);
  offset += 6;

  switch (stream_id + 0x100) {
    case ID_PS_PROGRAM_STREAM_MAP:
    case ID_PRIVATE_STREAM_2:
    case ID_ECM_STREAM:
    case ID_EMM_STREAM:
    case ID_PROGRAM_STREAM_DIRECTORY:
    case ID_DSMCC_STREAM:
    case ID_ITU_TREC_H222_TYPE_E_STREAM:
      gst_adapter_flush (adapter, offset);
      packet->PES_packet_data_buffer
          = gst_adapter_take_buffer (adapter, PES_packet_length);
      break;
    case ID_PADDING_STREAM:
      break;
    default:{
      guint flags1, flags2;
      flags1 = data[offset++];
      flags2 = data[offset++];
      packet->PES_header_data_length = data[offset++];

      if (packet->PES_header_data_length + offset > avail) {
        GST_DEBUG_OBJECT (adapter, "PES_header_data_length = %d > %d",
            packet->PES_header_data_length, avail - offset);
        goto invalid_length;
      }

      packet->PES_scrambling_control = (flags1 >> 4) & 3;
      packet->PES_priority = (flags1 >> 3) & 1;
      packet->data_alignment_indicator = (flags1 >> 2) & 1;
      packet->copyright = (flags1 >> 1) & 1;
      packet->original_or_copy = flags1 & 1;
      packet->PTS_DTS_flags = (flags2 >> 6) & 3;
      packet->ESCR_flag = (flags2 >> 5) & 1;
      packet->ES_rate_flag = (flags2 >> 4) & 1;
      packet->DSM_trick_mode_flag = (flags2 >> 3) & 1;
      packet->additional_copy_info_flag = (flags2 >> 2) & 1;
      packet->PES_CRC_flag = (flags2 >> 1) & 1;
      packet->PES_extension_flag = flags2 & 1;

      /* check sync bit */
      if ((flags1 & 0xC0) != 0x80)
        goto lost_sync;

      /* PTS_DTS_flags */
      if (flags2 & 0x80) {
        packet->PTS = ((data[offset] & 0xE) << (30 - 1))
            + ((data[offset + 1] & 0xFE) << (15 - 1))
            + ((data[offset + 2] & 0xFE) >> 1);
        offset += 3;
        if (flags2 & 0x40) {
          packet->DTS = ((data[offset] & 0xE) << (30 - 1))
              + ((data[offset + 1] & 0xFE) << (15 - 1))
              + ((data[offset + 2] & 0xFE) >> 1);
          offset += 3;
        }
      }
      /* ESCR_flag */
      if (flags2 & 0x20) {
        packet->ESCR = h222ts_demux_read_PCR (data + offset);
        offset += 6;
      }
      /* ES_rate_flag */
      if (flags2 & 0x10) {
        packet->ES_rate = ((data[offset] & 0x7F) << 15)
            + (data[offset + 1] << 7)
            + ((data[offset + 2] & 0xFE) >> 1);
        offset += 3;
      }
      /* DSM_trick_mode_flag */
      if (flags2 & 0x08) {
        packet->trick_mode_control = data[offset++];
      }
      /* additional_copy_info_flag */
      if (flags2 & 0x04) {
        packet->additional_copy_info = data[offset++] & 0x7F;
      }
      /* PES_CRC_flag */
      if (flags2 & 0x02) {
        packet->previous_PES_packet_CRC =
            (data[offset] << 8) + data[offset + 1];
        offset += 2;
      }
      /* PES_extension_flag */
      if (flags2 & 0x01) {
        guint8 flags = data[offset++];
        /* PES_private_data_flag */
        if (flags & 0x80) {
          /* FIXME: need to be process ? */
          offset += 128 / 8;
        }
        /* pack_header_field_flag */
        if (flags & 0x40) {
          /* FIXME: need this pack_header ? */
          guint8 length = data[offset++];
          if (G_UNLIKELY (length + offset > avail)) {
            GST_DEBUG_OBJECT (adapter, "pack_header_length = %d > %d",
                length, avail - offset);
            goto invalid_length;
          }
          offset += length;
        }
        /* program_packet_sequence_counter_flag */
        if (flags & 0x20) {
          packet->program_packet_sequence_counter = data[offset++] & 0x7F;
          packet->MPEG1_MPEG2_indicator = 1 & (data[offset] >> 6);
          packet->original_stuff_length = data[offset++] & 0x3F;
        }
        /* P-STD_buffer_flag */
        if (flags & 0x10) {
          packet->P_STD_buffer_scale = (data[offset] >> 5) & 1;
          packet->P_STD_buffer_size =
              ((data[offset] & 0x1F) << 8) + data[offset + 1];
          offset += 2;
        }
        /* PES_extension_flag_2 */
        if (flags & 0x01) {
          guint8 length = data[offset++] & 0x7F;
          if (G_UNLIKELY (length + offset > avail)) {
            GST_DEBUG_OBJECT (adapter, "PES_extension_2_length = %d > %d",
                length, avail - offset);
            goto invalid_length;
          }
          offset += length;
        }
      }

      /* end of PES_extension_flag */
      /* stuffing_byte */
      if (G_UNLIKELY (packet->original_stuff_length + offset > avail)) {
        GST_DEBUG_OBJECT (adapter, "original_stuff_length = %d > %d",
            packet->original_stuff_length, avail - offset);
        goto invalid_length;
      }
      offset += packet->original_stuff_length;

      /* PES_packet_data_byte */
      gst_adapter_flush (adapter, offset);
      packet->PES_packet_data_buffer
          = gst_adapter_take_buffer (adapter, avail - offset);
      break;
    }
  }                             /* end of switch (stream_id) */
  return GST_FLOW_OK;

lost_sync:
  GST_DEBUG_OBJECT (adapter, "lost sync");
  return GST_FLOW_LOST_SYNC;

invalid_length:
  return GST_FLOW_ERROR;
}

static GstPad *
h222ts_demux_create_source_pad (GstH222TSDemux * demux, guint16 PID)
{
  GstH222TSDemuxClass *demux_class = GST_H222TS_DEMUX_GET_CLASS (demux);
  guint stream_type = H222TS_STREAM (demux, PID)->stream_type;
  GstPad *pad = NULL;
  gchar *name = NULL;
  GstCaps *gencaps = NULL, *caps = NULL;

  GST_DEBUG_OBJECT (demux, "create src pad for stream_type 0x%04X", stream_type);

  GST_DEBUG_OBJECT (demux, "trace 1");
  switch (stream_type) {
      /*
       * defined by H222 spec
       */
    case ST_RESERVED:
      goto skip;
    case ST_VIDEO_MPEG1:
    case ST_VIDEO_MPEG2:
      caps = gst_caps_new_simple ("video/mpeg",
          "mpegversion", G_TYPE_INT, stream_type == ST_VIDEO_MPEG1 ? 1 : 2,
          "systemstream", G_TYPE_BOOLEAN, FALSE, NULL);
      break;
    case ST_AUDIO_MPEG1:
    case ST_AUDIO_MPEG2:
      caps =
          gst_caps_new_simple ("audio/mpeg", "mpegversion", G_TYPE_INT, 1,
          NULL);
      break;
    case ST_PRIVATE_SECTIONS:
      /* FIXME: whether be handle as PES ..? */
      goto skip;
    case ST_AUDIO_AAC:
      caps =
          gst_caps_new_simple ("audio/mpeg", "mpegversion", G_TYPE_INT, 4,
          NULL);
      break;
    case ST_VIDEO_MPEG4:
      caps = gst_caps_new_simple ("video/mpeg",
          "mpegversion", G_TYPE_INT, 4,
          "systemstream", G_TYPE_BOOLEAN, FALSE, NULL);
      break;
      /*
       * defined as reserved by H222 spec
       */
    case ST_VIDEO_H264:
      name = g_strdup_printf ("video_%04X", PID);
      caps = gst_caps_new_simple ("video/x-h264", NULL);
      break;
      /*
       * defined as User Private
       */
    case ST_BD_AUDIO_LPCM:
      caps = gst_caps_new_simple ("audio/x-lpcm", NULL);
      break;
    case ST_BD_AUDIO_AC3:
      caps = gst_caps_new_simple ("audio/x-ac3", NULL);
      break;
    case ST_BD_AUDIO_DTS:
      caps = gst_caps_new_simple ("audio/x-dts", NULL);
      break;
    default:
      break;
  }                             /* end of switch (stream->stream_type) */
  GST_DEBUG_OBJECT (demux, "trace 2");
  name = g_strdup_printf ("srcpid_%04X", PID);

  /* general caps for arbitrary elementary stream */
  gencaps = gst_caps_new_simple (GST_H222TS_MIME_TYPE_ES,
                                 "stream_type", G_TYPE_INT, (guint) PID, NULL);
  GST_DEBUG_OBJECT (demux, "trace 3");
  if (caps != NULL) {
    GST_DEBUG_OBJECT (demux, "trace 4");
    gst_caps_append (gencaps, caps);
  }
  GST_DEBUG_OBJECT (demux, "trace 5") ;
  pad = gst_pad_new_from_template (demux_class->tmpl_src, name);
  GST_DEBUG_OBJECT (demux, "trace 6");
  gst_pad_set_caps (pad, gencaps);
  GST_DEBUG_OBJECT (demux, "trace 7");
  g_free (name);
  gst_caps_unref (caps);
  gst_caps_unref (gencaps);
  GST_DEBUG_OBJECT (demux, "trace 8");

skip:
  return pad;
}

static GstFlowReturn
h222ts_demux_process_stream_impl (GstH222TSDemux * demux, guint16 PID)
{
  H222TSPIDContext *ctx = H222TS_PID_CONTEXT (demux, PID);
  H222TSStream *stream = H222TS_STREAM (demux, PID);
  H222TSPESPacket packet;
  GstFlowReturn ret = GST_FLOW_OK;

  GST_DEBUG_OBJECT (demux, "PES prepared PID %04x", PID);

  ret = h222ts_demux_parse_PES_packet (ctx->adapter, &packet);
  if (G_UNLIKELY (ret != GST_FLOW_OK)) {
    GST_DEBUG_OBJECT (demux, "PES packet parse failure PID %04X", PID);
    return ret;
  }

  GST_DEBUG_OBJECT (demux, "PES packet parse success");

  if (G_UNLIKELY (packet.PES_packet_data_buffer == NULL))
    return GST_FLOW_OK;

  GST_DEBUG_OBJECT (demux, "there is PES_packet_data");

  if (G_UNLIKELY (stream->pad == NULL)) {
    GST_DEBUG_OBJECT (demux, "pad is NULL, so try to create");
    /* setup source pad for the elementary stream */
    stream->pad = h222ts_demux_create_source_pad (demux, PID);
    if (G_UNLIKELY (stream->pad == NULL)) {
      GST_DEBUG_OBJECT (demux, "pad creating failure");
      return GST_FLOW_OK;
    }

    GST_DEBUG_OBJECT (demux, "pad set active");
    gst_pad_set_active (stream->pad, TRUE);
    gst_element_add_pad (GST_ELEMENT_CAST (demux), stream->pad);
  }

  GST_DEBUG_OBJECT (demux, "try to push");
  gst_pad_push (stream->pad, packet.PES_packet_data_buffer);
  return GST_FLOW_OK;
}

static GstFlowReturn
h222ts_demux_process_stream (GstH222TSDemux * demux, H222TSPacket * packet)
{
  H222TSPIDContext *ctx = H222TS_PID_CONTEXT (demux, packet->PID);
  GstFlowReturn ret = GST_FLOW_OK;
  GstBuffer *buffer = packet->data_buffer;

  if (G_UNLIKELY (buffer == NULL))
    return ret;

  /* clear garbage if payload_unit_start */
  if (packet->payload_unit_start_indicator) {
    h222ts_demux_process_stream_impl (demux, packet->PID);
    gst_adapter_clear (ctx->adapter);
  }

  /* push PES_packet_data_bytes */
  gst_adapter_push (ctx->adapter, buffer);

  /* check if adapter has enough data ,and if sync is gone */
  ret = h222ts_demux_has_enough_data_for_PES_packet (ctx->adapter);

  switch (ret) {
    case GST_FLOW_OK:
      /* now, we have prepared PES packet */
      ret = h222ts_demux_process_stream_impl (demux, packet->PID);
      break;
    case GST_FLOW_NEED_MORE_DATA:
      ret = GST_FLOW_OK;
      break;
    case GST_FLOW_LOST_SYNC:
      /* FIXME: what to do... */
      break;
    default:
      break;
  }

  return ret;
}

/*
 * section handlers
 */

static void
h222ts_demux_parse_section_header (const guint8 * data,
    H222TSSectionHeader * header)
{
  g_memset (header, 0, sizeof (H222TSSectionHeader));
  header->table_id = *data++;
  header->section_syntax_indicator = (*data) >> 7;
  header->section_length = 0x3FF & (((data[0] & 0xF) << 8) + data[1]);
  data += 2;
  header->program_number = (data[0] << 8) + data[1];    /* if in PAT, then transport_stream_id */
  data += 2;
  header->version_number = ((*data) >> 1) & 0x1F;
  header->current_next_indicator = (*data) & 1;
  data++;
  header->section_number = *data++;
  header->last_section_number = *data;
}

static void
h222ts_demux_dump_section_header (GstH222TSDemux * demux,
    H222TSSectionHeader * header)
{
  GST_DEBUG_OBJECT (demux, "section_header->table_id = 0x%02x",
      header->table_id);
  GST_DEBUG_OBJECT (demux, "section_header->section_syntax_indicator = %d",
      header->section_syntax_indicator);
  GST_DEBUG_OBJECT (demux, "section_header->section_length = 0x%04x",
      header->section_length);
  GST_DEBUG_OBJECT (demux, "section_header->program_number = 0x%04x",
      header->program_number);
  GST_DEBUG_OBJECT (demux, "section_header->version_number = 0x%02x",
      header->version_number);
  GST_DEBUG_OBJECT (demux, "section_header->section_number = 0x%02x",
      header->section_number);
  GST_DEBUG_OBJECT (demux, "section_header->last_section_number = 0x%02x",
      header->last_section_number);
}

static GstFlowReturn
h222ts_demux_has_enough_data_for_section (GstAdapter * adapter,
    H222TSSectionHeader * header)
{
  const guint8 *data;

  data = gst_adapter_peek (adapter, 8);
  if (G_UNLIKELY (data == NULL))
    return GST_FLOW_NEED_MORE_DATA;

  h222ts_demux_parse_section_header (data, header);

  if (G_UNLIKELY (header->section_length + 5 > gst_adapter_available (adapter)))
    return GST_FLOW_NEED_MORE_DATA;

  return GST_FLOW_OK;
}

static GstFlowReturn
h222ts_demux_process_PMT (GstH222TSDemux * demux, guint PID)
{
  H222TSSection *section = H222TS_SECTION (demux, PID);
  GstAdapter *adapter = H222TS_PID_CONTEXT (demux, PID)->adapter;
  guint program_info_length, offset = 0, avail;
  const guint8 *data;

  /* assume that section->header is fill,
     and that section_length is available in adapter */
  avail = section->header.section_length + 5;
  data = gst_adapter_peek (adapter, section->header.section_length + 5);
  offset += SECTION_HEADER_LENGTH;

  H222TS_PROGRAM (demux, section->header.program_number)->PCR_PID =
      ((data[offset] & 0x1F) << 8) + data[offset + 1];
  offset += 2;

  program_info_length = ((data[offset] & 3) << 8) + data[offset + 1];
  offset += 2;

  /* skip descriptor() */
  offset += program_info_length;

  while (offset + 5 + 4 <= avail) {
    guint8 stream_type = data[offset];
    guint16 elementary_PID =
        ((data[offset + 1] & 0x1F) << 8) + data[offset + 2];
    guint16 ES_info_length = ((data[offset + 3] & 0xF) << 8) + data[offset + 4];
    H222TSStream *stream = H222TS_STREAM (demux, elementary_PID);
    offset += 5;

    GST_DEBUG_OBJECT (demux, "[PMT entry] PID 0x%04X stream_type %04X",
        elementary_PID, stream_type);

    if (stream == NULL)
      stream = h222ts_demux_stream_new ();

    stream->stream_type = stream_type;
    H222TS_PID_CONTEXT_SET (demux, elementary_PID, stream);

    /* skip ES_info */
    offset += ES_info_length;
  }

  /* FIXME: should not ignore CRC_32... */

  return GST_FLOW_OK;
}

static GstFlowReturn
h222ts_demux_process_PAT (GstH222TSDemux * demux)
{
  H222TSSection *section =
      H222TS_SECTION (demux, PID_PROGRAM_ASSOCIATION_TABLE);
  GstAdapter *adapter = section->pid_context.adapter;
  guint offset = 0;
  guint num_of_programs;
  const guint8 *data;

  /* assume that section->header is fill,
     and that section_length is available in adapter */
  data = gst_adapter_peek (adapter, section->header.section_length + 5);
  offset += SECTION_HEADER_LENGTH;

  if (G_UNLIKELY (section->header.current_next_indicator == 0))
    return GST_FLOW_OK;

  num_of_programs = (section->header.section_length - 5 - 4) / 4;

  while (num_of_programs--) {
    gint program_number = (data[offset] << 8) + data[offset + 1];
    guint16 PID = ((data[offset + 2] & 0x1F) << 8) + data[offset + 3];
    offset += 4;

    GST_DEBUG_OBJECT (demux, "[PAT entry] program_number:0x%02X PID:0x%02X",
        program_number, PID);

    if (program_number == 0) {
      demux->network_PID = PID;
    } else {
      if (demux->program.program_number == -1) {
        demux->program.program_number = program_number;
        H222TS_PROGRAM (demux, program_number)->PMT_PID = PID;
        H222TS_PID_CONTEXT_SET (demux, PID, h222ts_demux_section_new ());
        GST_DEBUG_OBJECT (demux, "PMT PID is 0x%04X", PID);
      }
    }
  }

  /* FIXME: need to check CRC_32 */
  ;

  return GST_FLOW_OK;
}

static gboolean
h222ts_demux_PID_is_PMT (GstH222TSDemux * demux, guint PID)
{
  return demux->program.PMT_PID == PID;
}

static GstFlowReturn
h222ts_demux_process_section_impl (GstH222TSDemux * demux, guint PID)
{
  h222ts_demux_dump_section_header (demux, &H222TS_SECTION (demux,
          PID)->header);

  switch (PID) {

    case PID_PROGRAM_ASSOCIATION_TABLE:
      return h222ts_demux_process_PAT (demux);

#ifdef _SUPPORT_CAT
    case PID_CONDITIONAL_ACCESS_TABLE:
      return h222ts_demux_process_CAT (demux);
#endif
  }

  if (h222ts_demux_PID_is_PMT (demux, PID))
    return h222ts_demux_process_PMT (demux, PID);

  return TRUE;
}

static GstFlowReturn
h222ts_demux_process_section (GstH222TSDemux * demux, H222TSPacket * packet)
{
  H222TSSection *section = H222TS_SECTION (demux, packet->PID);
  GstAdapter *adapter = H222TS_PID_CONTEXT (demux, packet->PID)->adapter;
  GstBuffer *buffer = packet->data_buffer;

  if (G_UNLIKELY (buffer == NULL)) {
    GST_DEBUG_OBJECT (demux, "no data byte in PID_%04X", packet->PID);
    return GST_FLOW_OK;
  }

  if (packet->payload_unit_start_indicator) {
    GstBuffer *subbuf;
    guint8 pointer_field = *GST_BUFFER_DATA (buffer);
    if (G_UNLIKELY (1 + pointer_field > GST_BUFFER_SIZE (buffer))) {
      GST_DEBUG_OBJECT (demux, "invalid pointer_field %02X", pointer_field);
      gst_adapter_clear (adapter);
      return GST_FLOW_ERROR;
    }
    if (pointer_field) {
      /* push latest table segment, then parse the table */
      subbuf =
          gst_buffer_create_sub (buffer, 1, MIN (1 + pointer_field,
              GST_BUFFER_SIZE (buffer) - 1));
      gst_adapter_push (adapter, subbuf);
      if (G_UNLIKELY (GST_FLOW_OK ==
              h222ts_demux_has_enough_data_for_section (adapter,
                  &section->header)))
        h222ts_demux_process_section_impl (demux, packet->PID);
    }
    /* resize buffer for new table */
    GST_BUFFER_DATA (buffer) += 1 + pointer_field;
    GST_BUFFER_SIZE (buffer) -= 1 + pointer_field;
    /* flush garbage */
    gst_adapter_clear (adapter);
  }

  gst_adapter_push (adapter, buffer);
  if (G_LIKELY (GST_FLOW_OK ==
          h222ts_demux_has_enough_data_for_section (adapter, &section->header)))
    return h222ts_demux_process_section_impl (demux, packet->PID);

  return GST_FLOW_OK;
}

/*
 * transport packet handlers
 */

static gint
h222ts_demux_parse_adaptation_field (GstBuffer * buffer, H222TSPacket * packet)
{
  guint8 adaptation_field_length;
  guint8 flags;
  guint offset = 0;
  const guint8 *data = GST_BUFFER_DATA (buffer);

  adaptation_field_length = data[offset++];

  if (G_UNLIKELY (adaptation_field_length > GST_BUFFER_SIZE (buffer))) {
    GST_DEBUG_OBJECT (buffer, "invalid adaptation_field_length 0x%02X",
        adaptation_field_length);
    goto failure;
  }

  if (G_UNLIKELY (adaptation_field_length == 0))
    goto done;

  flags = data[offset++];
  packet->discontinuity_indicator = 1 & (flags >> 7);
  packet->random_access_indicator = 1 & (flags >> 6);
  packet->elementary_stream_priority_indicator = 1 & (flags >> 5);

  /* PCR_flag */
  if (flags & 0x10) {
    packet->PCR_flag = 1;
    packet->PCR = h222ts_demux_read_PCR (data + offset);
    offset += 6;
  }
  /* OPCR_flag */
  if (flags & 0x08) {
    packet->OPCR = h222ts_demux_read_PCR (data + offset);
    offset += 6;
  }
  /* splicing_point_flag */
  if (flags & 0x04) {
    packet->splice_countdown = data[offset++];
  }
  /* transport_private_data_flag */
  if (flags & 0x02) {
    guint8 length = data[offset++];

    if (length > GST_BUFFER_SIZE (buffer) - offset) {
      GST_DEBUG_OBJECT (buffer, "invalid transport_private_data_length 0x%02X",
          length);
      goto failure;
    }

    packet->private_data_buffer =
        gst_buffer_create_sub (buffer, offset, length);
    offset += length;
  }
  /* adaptation_field_extension_flag */
  if (flags & 0x01) {
    guint8 length = data[offset++];
    guint8 flags2, skip_to = offset + length;

    if (length > GST_BUFFER_SIZE (buffer) - offset) {
      GST_DEBUG_OBJECT (buffer,
          "invalid adaptation_field_extension_length 0x%02X", length);
      goto failure;
    }

    flags2 = data[offset++];

    /* ltw_flag */
    if (flags2 & 0x80) {
      packet->ltw_valid_flag = (data[offset] >> 7) & 1;
      packet->ltw_offset = ((data[offset] << 8) & ~0x80) + data[offset + 1];
      offset += 2;
    }
    /* piecewise_rate_flag */
    if (flags2 & 0x40) {
      packet->piecewise_rate =
          ((data[offset] & ~0xC0) << 16) + (data[offset + 1] << 8) +
          data[offset + 2];
      offset += 3;
    }
    /* seamless_splice_flag */
    if (flags2 & 0x20) {
      packet->splice_type = (data[offset] & 0xF0) >> 4;
      packet->DTS_next_AU =
          ((data[offset] & 0x0E) << 30) + ((data[offset + 1] & ~0x01) << 15) +
          ((data[offset + 2] & ~0x01) >> 1);
      offset += 3;
    }
    /* skip reserved bytes */
    if (skip_to < offset) {
      GST_DEBUG_OBJECT (buffer, "skip_to < offset");
      goto failure;
    }
    offset = skip_to;
  }

done:
  return adaptation_field_length;

failure:
  return -1;
}

static void
h222ts_demux_dump_transport_packet (GstObject * demux, H222TSPacket * packet)
{
  GST_DEBUG_OBJECT (demux, "transport packet ->transport_error_indicator = %d",
      packet->transport_error_indicator);
  GST_DEBUG_OBJECT (demux,
      "transport packet ->payload_unit_start_indicator = %d",
      packet->payload_unit_start_indicator);
  GST_DEBUG_OBJECT (demux, "transport packet ->transport_priority = %d",
      packet->transport_priority);
  GST_DEBUG_OBJECT (demux, "transport packet ->PID = 0x%04x", packet->PID);
  GST_DEBUG_OBJECT (demux,
      "transport packet ->transport_scrambling_control = %d",
      packet->transport_scrambling_control);
  GST_DEBUG_OBJECT (demux, "transport packet ->adaptation_field_control = %d",
      packet->adaptation_field_control);
  GST_DEBUG_OBJECT (demux, "transport packet ->continuity_counter = %d",
      packet->continuity_counter);
}

static GstFlowReturn
h222ts_demux_parse_transport_packet (GstBuffer * buffer, H222TSPacket * packet)
{
  const guint8 *data = GST_BUFFER_DATA (buffer);
  guint offset = 0;

  /* check sync byte */
  if (data[offset++] != 0x47) {
    GST_DEBUG_OBJECT (buffer, "lost sync");
    return GST_FLOW_LOST_SYNC;
  }

  /* parse static transport packet header */
  packet->transport_error_indicator = (data[offset] & 0x80) == 0x80;
  packet->payload_unit_start_indicator = (data[offset] & 0x40) == 0x40;
  packet->transport_priority = (data[offset] & 0x20) == 0x20;
  packet->PID = ((data[offset] & 0x1F) << 8) + data[offset + 1];
  offset += 2;
  packet->transport_scrambling_control = (data[offset] & 0xc0) >> 6;
  packet->adaptation_field_control = (data[offset] & 0x30) >> 4;
  packet->continuity_counter = data[offset] & 0x0f;
  offset++;

  h222ts_demux_dump_transport_packet (GST_OBJECT (buffer), packet);

  /* parse adaptation_field if any */
  if (packet->adaptation_field_control & 2) {
    GstBuffer *subbuf =
        gst_buffer_create_sub (buffer, offset,
        GST_BUFFER_SIZE (buffer) - offset);
    gint length = h222ts_demux_parse_adaptation_field (subbuf, packet);

    if (length < 0)
      return GST_FLOW_ERROR;

    offset += length;
  }

  /* take data_byte as GstBuffer if any */
  if (packet->adaptation_field_control & 1) {
    packet->data_buffer =
        gst_buffer_create_sub (buffer, offset,
        GST_BUFFER_SIZE (buffer) - offset);
  }

  return GST_FLOW_OK;
}

static void
h222ts_demux_program_update_PCR (GstH222TSDemux * demux, guint program_number,
    guint64 pcr)
{
}

static gboolean
h222ts_demux_is_PCR_PID (GstH222TSDemux * demux, guint16 PID)
{
  return demux->program.PCR_PID == PID;
}

static gboolean
h222ts_demux_process_packet (GstH222TSDemux * demux, H222TSPacket * packet)
{
  H222TSPIDContext *ctx;

  /* check NULL packet */
  if (packet->PID == PID_NULL_PACKET) {
    GST_DEBUG_OBJECT (demux, "NULL packet");
    return TRUE;
  }
  /* check if PID is registered */
  ctx = H222TS_PID_CONTEXT (demux, packet->PID);
  if (G_UNLIKELY (ctx == NULL)) {
    GST_DEBUG_OBJECT (demux, "PID_%04X is not registered", packet->PID);
    return FALSE;
  }
  /* check duplication */
  if (ctx->continuity_counter == packet->continuity_counter) {
    GST_DEBUG_OBJECT (demux, "duplicatiion found in PID_%04X", packet->PID);
    /* update only PCR if present */
    if (packet->PCR_flag && H222TS_PROGRAM (demux, ctx->program_number) != NULL)
      h222ts_demux_program_update_PCR (demux, ctx->program_number, packet->PCR);
    return TRUE;
  }
  /* check discontinuity */
  if (packet->discontinuity_indicator) {
    if (h222ts_demux_is_PCR_PID (demux, packet->PID)) {
      /* system time-base discontinuities */
      GST_DEBUG_OBJECT (demux, "system time-base discontinuities");
      h222ts_demux_program_update_PCR (demux, ctx->program_number, packet->PCR);
    } else {
      /* continuity_counter discontinuities */
      /* this is processed by packet specific processor */
      GST_DEBUG_OBJECT (demux, "continuity_counter discontinuities");
    }
  } else if (ctx->continuity_counter != -1 &&
      packet->continuity_counter != (ctx->continuity_counter + 1) % (1 << 4)) {
    GST_DEBUG_OBJECT (demux,
        "discontinuity_indicator==0, but continuity_counter discontinuities..");
    return FALSE;
  }

  /* packet specific process */
  GST_DEBUG_OBJECT (demux, "process PID 0x%04X", packet->PID);
  return ctx->process != NULL ? ctx->process (demux, packet) : FALSE;
}

static GstFlowReturn
h222ts_demux_take_transport_packet_buffer (GstAdapter * adapter,
    GstBuffer ** buffer)
{
  guint i, avail = gst_adapter_available (adapter);
  const guint8 *data = gst_adapter_peek (adapter, avail);

  GST_DEBUG_OBJECT (adapter, "prepare a transport packet");

  *buffer = NULL;

  if (G_UNLIKELY (avail < H222TS_NORMAL_TS_PACKETSIZE))
    return GST_FLOW_NEED_MORE_DATA;

  for (i = 0; i < avail; i++)
    if (data[i] == 0x47) {
      GST_DEBUG_OBJECT (adapter, "detect sync [%02X] at offset %d", data[i], i);
      break;
    }

  if (i)
    gst_adapter_flush (adapter, i);

  if (gst_adapter_available (adapter) < H222TS_NORMAL_TS_PACKETSIZE)
    return GST_FLOW_NEED_MORE_DATA;

  *buffer = gst_adapter_take_buffer (adapter, H222TS_NORMAL_TS_PACKETSIZE);
  return *buffer == NULL ? GST_FLOW_NEED_MORE_DATA : GST_FLOW_OK;
}

static GstBuffer *
h222ts_demux_take_packet_buffer (GstH222TSDemux * demux)
{
  GstH222TSDemuxClass *demux_class = GST_H222TS_DEMUX_CLASS (demux);
  GstAdapter *adapter = demux->adapter;
  GstBuffer *buffer = NULL;

  h222ts_demux_take_transport_packet_buffer (adapter, &buffer);
  return buffer;

  if (demux_class->take_transport_packet_buffer) {
    demux_class->take_transport_packet_buffer (adapter, &buffer);
  } else if (gst_adapter_available (adapter) >= H222TS_NORMAL_TS_PACKETSIZE) {
    buffer = gst_adapter_take_buffer (adapter, H222TS_NORMAL_TS_PACKETSIZE);
  }
  return buffer;
}

static GstFlowReturn
h222ts_demux_sink_chain (GstPad * pad, GstBuffer * buffer)
{
  GstH222TSDemux *demux = GST_H222TS_DEMUX (gst_pad_get_parent (pad));
  H222TSPacket packet;
  GstBuffer *tsp_buf = NULL;
  GstFlowReturn ret = GST_FLOW_OK;

  GST_DEBUG_OBJECT (demux, "detect sink buffer");

  gst_adapter_push (demux->adapter, buffer);

  while (NULL != (tsp_buf = h222ts_demux_take_packet_buffer (demux))) {

    GST_DEBUG_OBJECT (demux, "detect a transport packet");
    g_memset (&packet, 0, sizeof (H222TSPacket));

    ret = h222ts_demux_parse_transport_packet (tsp_buf, &packet);
    if (G_LIKELY (ret == GST_FLOW_OK))
      h222ts_demux_process_packet (demux, &packet);
  }

  return GST_FLOW_OK;
}
